package org.apache.hadoop.hdfs.util;

import static org.apache.hadoop.util.Time.monotonicNow;

/** 
 * a class to throttle the data transfers.
 * This class is thread safe. It can be shared by multiple threads.
 * The parameter bandwidthPerSec specifies the total bandwidth shared by
 * threads.
 */
public class DataTransferThrottler {
  private final long period;          // period over which bw is imposed
  private final long periodExtension; // Max period over which bw accumulates.
  private long bytesPerPeriod;  // total number of bytes can be sent in each period
  private long curPeriodStart;  // current period starting time
  private long curReserve;      // remaining bytes can be sent in the period
  private long bytesAlreadyUsed;

  /** Constructor 
   * @param bandwidthPerSec 允许的带宽（字节/秒）.
   */
  public DataTransferThrottler(long bandwidthPerSec) {
    this(500, bandwidthPerSec);  // 默认节流周期为500ms
  }

  /**
   * Constructor
   * @param period in milliseconds. Bandwidth is enforced over this period.
   * @param bandwidthPerSec bandwidth allowed in bytes per second. 
   */
  public DataTransferThrottler(long period, long bandwidthPerSec) {
    this.curPeriodStart = monotonicNow();
    this.period = period;
    this.curReserve = this.bytesPerPeriod = bandwidthPerSec * period / 1000; // 1024*1024 *500 / 1000
    // 每毫秒传输的字节数量，大概是每秒允许传输10kb的数据
    // 然后再把每毫秒允许传输的数据 * 500 （period） = 它会对每500毫秒传输量进行一个计数，如果每500毫秒时间窗口的数据传输量超过了设置的阈值，就会等待，不让继续传输数据。
    this.periodExtension = period * 3;  // 1500ms
  }

  /**
   * @return current throttle bandwidth in bytes per second.
   */
  public synchronized long getBandwidth() {
    return bytesPerPeriod*1000/period;
  }
  
  /**
   * Sets throttle bandwidth. This takes affect latest by the end of current
   * period.
   */
  public synchronized void setBandwidth(long bytesPerSecond) {
    if ( bytesPerSecond <= 0 ) {
      throw new IllegalArgumentException("" + bytesPerSecond);
    }
    bytesPerPeriod = bytesPerSecond*period/1000;
  }
  
  /** Given the numOfBytes sent/received since last time throttle was called,
   * make the current thread sleep if I/O rate is too fast
   * compared to the given bandwidth.
   *
   * @param numOfBytes number of bytes sent/received since last time throttle was called
   */
  public synchronized void throttle(long numOfBytes) {
    throttle(numOfBytes, null);
  }

  /** Given the numOfBytes sent/received since last time throttle was called,
   * make the current thread sleep if I/O rate is too fast
   * compared to the given bandwidth.  Allows for optional external cancelation.
   *
   * @param numOfBytes number of bytes sent/received since last time throttle was called
   * @param canceler optional canceler to check for abort of throttle
   */
  public synchronized void throttle(long numOfBytes, Canceler canceler) {
    if ( numOfBytes <= 0 ) {
      return;
    }

    curReserve -= numOfBytes;  // 在数据传输的过程中，每次接收到一批数据，就得调用一下限流的算法
    // 看一下当前是否需要限流，当前这个500ms的时间窗口，还允许传输多少数据
    // 每次你传输了一部分数据之后，就会把500ms的时间窗口内允许传输的数据 减去 已经传输的数据
    bytesAlreadyUsed += numOfBytes; // 累加 500ms内 已经传输了多少数据

    while (curReserve <= 0) {
      if (canceler != null && canceler.isCancelled()) {
        return;
      }
      long now = monotonicNow();
      long curPeriodEnd = curPeriodStart + period; // 假设限流组件初始化的时间： 20：00：00
      // 传输了一会儿数据之后，就触发了这个 curReserve <= 0 的条件，即 限定的流量阈值达标了
      // curPeriodEnd = 20:00:00 + 500ms

      // 按照 period 指定的时间限流，第一个时间窗口是 20:00:00 + 500ms
      if ( now < curPeriodEnd ) {
        // 这个限流组件初始化之后 500ms内，传输的数据量已经超过了设置的每秒多少个字节的量了
        try {
          wait( curPeriodEnd - now );
          // 在一个500ms的时间窗口内触发了限流，就需要让当前线程等待一会，
          // 不能继续传输数据，避免在一个时间窗口内传输的数据流太大
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          break;
        }
      } else if ( now <  (curPeriodStart + periodExtension)) {
        // now : 20:00:00 + 200ms
        // curPeriodStart: 20:00:00
        // periodExtension : 1500ms
        // now < 20:00:00 + 1500ms
        curPeriodStart = curPeriodEnd; // curPeriodStart = 20:00:00 + 500ms
        curReserve += bytesPerPeriod; // curReserve
        // 切换成下一个时间窗口，可以继续传输了
      } else {
        curPeriodStart = now;
        curReserve = bytesPerPeriod - bytesAlreadyUsed;
      }
    }
    bytesAlreadyUsed -= numOfBytes;

    // throttle方法退出，负责数据传输的西安测绘给你就可以继续传输数据了
    // 但是一样的，每次传输了一部分数据之后，就会使用传输的数据量来调用throttle方法进行限流判断

    /**
     * 总结：
     * 1. 每次传输一点数据，就拿着传输的数据量来调用 throttle限流算法
     * 2. 不断的扣减时间窗口内允许传输的剩余数据量，5mb
     * 3. 如果时间窗口内剩余数据量小于等于0，就说明已经触发阈值了
     * 4. 判断当前时间是否在最近一个500ms的时间窗口内，如果是，则限流，wait， 不让线程继续传输数据
     * 5. 线程等待一段时间过后，会重新开启一个新的时间窗口，继续传输数据。
     */
  }
}
